package com.ibm.cps.kafka;

import org.apache.log4j.Logger;

import com.google.common.base.Throwables;
import com.ibm.util.exception.CPSException;

public abstract class KafkaTopicConsumeThread extends KafkaTopicConsumeTemplate implements Runnable {

	private Logger logger;

	public KafkaTopicConsumeThread(String zookeeper, String topicName) {
		super(zookeeper, topicName);
		logger = Logger.getLogger(KafkaTopicConsumeThread.class);
	}

	public KafkaTopicConsumeThread(String zookeeper, String groupid, String topicName) {
		super(zookeeper, groupid, topicName);
		logger = Logger.getLogger(KafkaTopicConsumeThread.class);
	}

	@Override
	public void run() {
		try {
			consumeMessages();
		} catch (CPSException e) {
			logger.error(Throwables.getStackTraceAsString(e));
		}
	}

}
